package com.foreveross.mofang.datapersistence.infra;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import javax.inject.Named;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.foreveross.mofang.datapersistence.infra.exception.HbaseOperationException;
import com.foreveross.mofang.datapersistence.infra.service.impl.DataPersistenceMsgServiceImpl;
import com.foreveross.mofang.datapersistence.infra.utils.HbaseManagerUtil;

/**
 * hbase仓储类实现
 * 
 * @author fb
 */
@Named
public class HbaseRepositoryImpl implements IHbaseRepository {

	private static final Logger logger = LoggerFactory.getLogger(DataPersistenceMsgServiceImpl.class);

	public void createTable(String tableName, List<String> familys) throws Exception {
		try {
			HbaseManagerUtil.getInstance().createTable(tableName, familys);
		} catch (Exception e) {
			logger.error("创建HbaseManagerUtil实际时出错:", e);
			throw new HbaseOperationException(e);
		}
	}

	public void addRecord(String tableName, String rowKey , Map<String, Object> rowData) throws IOException {
		HTableInterface table = null;
		Put put = createPut(tableName, rowKey, rowData);
		
		if (put == null) {
			return;
		}

		try {
			table = HbaseManagerUtil.getInstance().getHtable(tableName);
			table.put(put);
		} finally {
			close(tableName);
		}
	}
	
	public void batchAddRecord(String tableName, Map<String, Map<String, Object>> rowDatas) throws IOException {
		HTableInterface table = HbaseManagerUtil.getInstance().getHtable(tableName);
		table.setAutoFlushTo(false);
		Put put = null;
		
		try {
			table = HbaseManagerUtil.getInstance().getHtable(tableName);
			
			for (Entry<String, Map<String, Object>> entry : rowDatas.entrySet()) {
				put = createPut(tableName, entry.getKey(), entry.getValue());
				
				if (put == null) {
					continue;
				}
				
				table.put(put);
			}
		} finally {
			close(tableName);
		}
	}

    public void delOneRecord (String tableName, String rowKey) throws Exception{
    	if (StringUtils.isBlank(rowKey)) {
    		logger.debug("hbase删除一行记录表名或rowkey为空，tableName:{}  rowkey:{}", tableName, rowKey);
    		
        	return;
    	}
    	
    	delMultRecord(tableName, rowKey);
    }
    
    public void delMultRecord (String tableName, String... rowKeys) throws Exception{
    	List<Delete> deletes = new ArrayList<Delete>();
    	
    	if (rowKeys == null || rowKeys.length < 1 || StringUtils.isBlank(tableName)) {
    		logger.debug("执行hbase 删除多行记录rowkeys或表名为空 rowkeys:{}", tableName, Arrays.toString(rowKeys));
    		
			return;
		}
    	
    	try{
        	for(String key : rowKeys){
    			deletes.add(new Delete(Bytes.toBytes(key.trim())));
    		}
        	
        	HbaseManagerUtil.getInstance().getHtable(tableName).delete(deletes);
    	} finally{
    		close(tableName);
        }
    }
     
    public Map<String, Object> getRecord(String tableName, String rowKey) throws Exception{
    	Map<String, Object> record = new HashMap<String, Object>();
    	HTableInterface table = null;
    	Get get = null;
    	Result rs = null;
    	
    	if (StringUtils.isBlank(rowKey) || StringUtils.isBlank(tableName)) {
			logger.debug("表名或rowkey为空，tableName:{}  rowkey:{}", tableName, rowKey);
			
			return null;
		}

    	try{
    		table = HbaseManagerUtil.getInstance().getHtable(tableName);
			get = new Get(Bytes.toBytes(rowKey));
			rs = table.get(get);
			
			for(Cell cell : rs.rawCells()){
				record.put(new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)));
		    }
			
			if (!record.isEmpty()) {
				record.put("rowKey", rowKey);
			}
    	} finally{
    		close(tableName);
        }
    	
    	return record;
    }
     
	public List<Map<String, Object>> queryRecords(String tableName, FilterList filterList ) throws Exception {
		List<Map<String, Object>> results = null;
		HTableInterface table = null;
		Scan scan = null;
		ResultScanner resultScanner = null;

		results = new ArrayList<Map<String, Object>>();
		scan = new Scan();

		if (filterList != null) {
			scan.setFilter(filterList);
		}

		try {
			table = HbaseManagerUtil.getInstance().getHtable(tableName);
			resultScanner = table.getScanner(scan);
			for (Result rs : resultScanner) {
				Map<String, Object> record = new HashMap<String, Object>();
				record.put("rowKey", new String(rs.getRow()));

				for (Cell cell : rs.rawCells()) {
					record.put(new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)));
				}

				results.add(record);
			}

		} finally {
			close(tableName);
		}

		return results;
	}
	
	@SuppressWarnings("unchecked")
	private Put createPut(String tableName, String rowKey,
			Map<String, Object> rowData) {
		String[] familyQual  = null;
		Put put = new Put(Bytes.toBytes(rowKey));
		
		if (StringUtils.isBlank(rowKey) || StringUtils.isBlank(tableName)) {
			logger.warn("hbase插入表名或rowkey为空，tableName:{}  rowkey:{}", tableName, rowKey);
			
			return null;
		}
		
		for (Entry<String, Object> entry : rowData.entrySet()) {
			if (entry.getValue() == null) {
				continue;
			}
			
			if (entry.getValue() instanceof Map) {
				for (Entry<String, String> chlid : ((Map<String, String>) entry.getValue()).entrySet()) {
					if (StringUtils.isBlank(chlid.getValue())) {
						continue;
					}
					
					familyQual = entry.getKey().split(Constant.KEY_SEPARATOR);
					put.add(Bytes.toBytes(familyQual[0]),
							Bytes.toBytes(familyQual[1]),
							Bytes.toBytes(chlid.getValue()));
				}
			} else {
				if (StringUtils.isBlank(entry.getValue().toString())) {
					continue;
				}
				
				familyQual = entry.getKey().split(Constant.KEY_SEPARATOR);
				put.add(Bytes.toBytes(familyQual[0]),
						Bytes.toBytes(familyQual[1]),
						Bytes.toBytes(entry.getValue().toString()));
			}
		}
		return put;
	}
 
    private void close(String tableName) {
    	try {
			HbaseManagerUtil.getInstance().releaseHTable(tableName);
		} catch (IOException e) {
			logger.warn("hbase {}表关闭异常： ", tableName, e);
		}
    }
}
